#include "MessageQueue.h"
#include "utils/utils.h"

namespace zeroplus::mq
{
    MessageQueue::MessageQueue(std::shared_ptr<sw::redis::Redis> redisFd)
        : redis_(redisFd),
          messageOffset_("")
    {
    }

    MessageQueue::~MessageQueue()
    {
    }

    std::string MessageQueue::CreateStreams(const std::string &topic)
    {
        if (!IsTopicExists({topic}))
        {
            std::vector<std::pair<std::string, std::string>> attrs = {{"message", "MessageQueue"}};
            redis_->xadd(topic, "*", attrs.begin(), attrs.end());
        }
        return "";
    }

    std::string MessageQueue::WriteToStorage(const std::string &topic, const Bytes &message)
    {
        std::vector<std::pair<std::string, std::string>> attrs = {{"message", message.to_string()}};
        return redis_->xadd(topic, "*", attrs.begin(), attrs.end());
    }

    std::tuple<Bytes, std::string, bool> MessageQueue::ReadFromStorage(const std::string &topic, const std::string &consumerGroup, const std::string &consumerId,
                                                                       const int &count, const int &blockSeconds)
    {
        // LOG_DEBUG("读取数据, 参数： topic=%s, consumerGroup=%s, consumerId=%s", topic.c_str(), consumerGroup.c_str(), consumerId.c_str());
        std::unordered_map<std::string, ItemStream> result;
        while (messageOffset_.empty())
        {
            redis_->xread(topic, "0-0", 1, std::inserter(result, result.end()));
            auto it = std::find_if(result.begin(), result.end(), [](const auto &pair)
                                   { return !pair.second.empty(); });
            if (it != result.end())
            {
                messageOffset_ = it->second.front().first;
            }
        }
        if (consumerGroupMap_.find(consumerGroup) == consumerGroupMap_.end())
        {
            try
            {
                LOG_DEBUG("创建消费者组");
                redis_->xgroup_create(topic, consumerGroup, messageOffset_, false);
            }
            catch (const std::exception &e)
            {
                LOG_WARN("消费者组%s已存在于redis中 %s", consumerGroup.c_str(), e.what());
            }
            consumerGroupMap_[consumerGroup] = true;
        }
        result.clear();
        redis_->xreadgroup(consumerGroup, consumerId, topic, ">", std::chrono::milliseconds{blockSeconds},
                           count, false, std::inserter(result, result.end()));

        for (const auto &[stream, items] : result)
        {
            for (const auto &[id, attrs] : items)
            {
                if (attrs)
                {
                    for (const auto &[field, value] : *attrs)
                    {
                        return std::make_tuple(Bytes::from_string(value), id, true);
                    }
                }
                else
                {
                    LOG_ERROR("No attributes for this message.");
                    return std::make_tuple(Bytes::create(0), "", false);
                }
            }
        }
        return std::make_tuple(Bytes::create(0), "", false);
    }

    std::tuple<Bytes, std::string, bool> MessageQueue::ReadFromStorage(const std::string &topic, const std::string &messageId, const int &count)
    {
        // LOG_DEBUG("读取数据, 参数： topic=%s, consumerGroup=%s, consumerId=%s", topic.c_str(), consumerGroup.c_str(), consumerId.c_str());
        while (messageOffset_.empty())
        {
            std::unordered_map<std::string, ItemStream> result;
            redis_->xread(topic, "0-0", 1, std::inserter(result, result.end()));
            for (const auto &[stream, items] : result)
            {
                for (const auto &[id, attrs] : items)
                {
                    LOG_DEBUG("MessageId: %s", id.c_str())
                    messageOffset_ = id;
                }
            }
        }

        ItemStream itemStream;
        redis_->xrange(topic, messageId, messageId, std::back_inserter(itemStream));
        // 解析并输出结果
        for (const auto &[id, attrs] : itemStream)
        {
            if (attrs)
            {
                for (const auto &[field, value] : *attrs)
                {
                    return std::make_tuple(Bytes::from_string(value), id, true);
                }
            }
            else
            {
                LOG_ERROR("No attributes for this message.");
                return std::make_tuple(Bytes::create(0), "", false);
            }
        }
    }

    // 获取指定队列的大小
    long long MessageQueue::Length(const std::string &topic)
    {
        return redis_->xlen(topic);
    }

    std::vector<std::string> MessageQueue::GetAllTopic()
    {
        std::vector<std::string> streamKeys;
        sw::redis::Cursor cursor = 0;
        std::unordered_set<std::string> keys;
        do
        {
            cursor = redis_->scan(cursor, "*", 10, std::inserter(keys, keys.begin()));
        } while (cursor != 0);

        for (const auto &key : keys)
        {
            auto keyType = redis_->type(key);
            if (keyType == "stream")
            {
                streamKeys.push_back(key);
            }
        }
        return streamKeys;
    }

    long long MessageQueue::IsTopicExists(const std::vector<std::string> &topicList)
    {
        return redis_->exists(topicList.begin(), topicList.end());
    }

    bool MessageQueue::Acknowledge(const std::string &topic, const std::string &messageId, const std::string &groupName)
    {
        auto ackCnt = redis_->xack(topic, groupName, messageId);
        LOG_DEBUG("消息已应答: %s", messageId.c_str());
        auto delCnt = redis_->xdel(topic, messageId);
        return delCnt && ackCnt;
    }

    std::vector<PendingItem> MessageQueue::SerachPendingMessage(const std::string &topic, const std::string &consumerGroup, const std::string &consumerId, const std::string &messageId)
    {
        std::vector<PendingItem> pendingItems;
        redis_->xpending(topic, consumerGroup, messageId, messageId, 1, consumerId, std::back_inserter(pendingItems));
        return pendingItems;
    }

    std::vector<ClaimedItem> MessageQueue::ClaimPendingMessage(const std::string &topic, const std::string &consumerGroup, const std::string &consumerId, const std::string &messageId)
    {
        std::vector<ClaimedItem> claimedItems;
        redis_->xclaim(topic, consumerGroup, consumerId, std::chrono::milliseconds(0),
                       messageId, std::back_inserter(claimedItems));
        return claimedItems;
    }
}
